package com.bestcem.xm.common.core.config.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamRetryTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Base64;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @ClassName: StreamConfig
 * @Description:
 * @Author: chenglong.yue
 * @Date: 2022/5/3 14:31
 */
@Slf4j
@Configuration
@EnableAsync
public class XmCustomStreamConfig {

    @Bean
    public MessageConverter customMessageConverter() {
        return new XmCustomMessageConverter();
    }

    @Bean(name = "mqDelayMqExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
        pool.setKeepAliveSeconds(30);
        //核心线程池数
        pool.setCorePoolSize(1);
        //最大线程
        pool.setMaxPoolSize(10);
        //队列容量
        pool.setQueueCapacity(1024);
        //队列满，线程被拒绝执行策略
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        pool.setThreadNamePrefix("mq-delay-pool-");
        return pool;
    }

    @StreamRetryTemplate
    public RetryTemplate xmDefaultRetryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        // 设置重试回退操作策略，主要设置重试间隔时间
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        // 间隔2s重试
        backOffPolicy.setBackOffPeriod(2000);

        //设置重试次数和异常类型，1次代表就是不重试
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(1);

        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        return retryTemplate;
    }

    /*@StreamRetryTemplate
    public RetryTemplate memberRetryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        // 设置重试回退操作策略，主要设置重试间隔时间
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        // 间隔5s重试
        backOffPolicy.setBackOffPeriod(5000);

        //设置重试次数和异常类型，1次代表就是不重试
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(3);

        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        return retryTemplate;
    }*/

    @ServiceActivator(inputChannel = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
    //@ServiceActivator(inputChannel = "errorChannel")
    public void messageException(ErrorMessage errorMessage) {
        Throwable throwable = errorMessage.getPayload();
        Message<?> originalMessage = errorMessage.getOriginalMessage();
        if (Objects.isNull(originalMessage)) {
            return;
        }
        String message = Base64.getEncoder().encodeToString((byte[]) originalMessage.getPayload());
        MessageHeaders headers = originalMessage.getHeaders();
        log.error("消息处理失败,headers:{}, message:{}", headers, message, throwable);
    }
}
